package com.bw.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Random;

public class ProcessingPage {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> dataStreamSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log_project", "dwd_traffic_page_log_project_1"));
//        dataStreamSource.print();
        SingleOutputStreamOperator<JSONObject> map = dataStreamSource.map(t -> JSONObject.parseObject(t));
        KeyedStream<JSONObject, String> jsonObjectStringKeyedStream = map.keyBy(t -> t.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> process = jsonObjectStringKeyedStream.process(new ProcessFunction<JSONObject, JSONObject>() {
            ValueState<String> lastValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                lastValueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastValueState", String.class));
            }

            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                String string = value.getJSONObject("page").getString("last_page_id");
                String string1 = value.getJSONObject("page").getString("page_id");
                if (string == null) {
                    Random random = new Random();
                    int i = random.nextInt(10);
                    lastValueState.update("shop_" + (i + 1));
                        value.getJSONObject("page").put("shop_id", "shop_" + (i + 1));
                } else {
                    if (lastValueState.value() == null) {
                        Random random = new Random();
                        int i = random.nextInt(10);
                        lastValueState.update("shop_" + (i + 1));

                            value.getJSONObject("page").put("shop_id", "shop_" + (i + 1));
                    } else {
                            value.getJSONObject("page").put("shop_id", lastValueState.value());

                    }
                }
                out.collect(value);
            }
        });
        process.print();
        process.map(t -> t.toJSONString()).addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_page_log_project_shop"));
        env.execute();
    }
}
